use std::any::Any;
use tokio::net::TcpListener;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::pin::Pin;
use std::error::Error;
use std::future::Future;
use std::io::Read;
use std::marker::PhantomData;
use std::net::Shutdown;
use std::sync::Arc;
use std::thread::sleep;
use std::time::Duration;
use anyhow::{anyhow, bail};
use log::debug;
use serde_json::{json, Value};
use tokio::net::TcpStream;
use tokio::sync::Mutex;
use tokio::time::interval;
use tokio::io::{AsyncWriteExt,AsyncReadExt};
use tokio::{runtime};
use tokio::net::windows::named_pipe::PipeEnd::Server;

// 定义 JSON - RPC 响应结构体
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct JsonRpcPacket {
    pub jsonrpc: String,
    pub method: Option<String>,
    pub result: Option<Value>,
    pub error: Option<Value>,
    pub params: Option<Vec<Value>>,
    pub id: Option<i32>,
}

impl JsonRpcPacket {
    pub(crate) fn new() -> JsonRpcPacket {
        JsonRpcPacket{
            jsonrpc: "2.0".to_string(),
            method: None,
            result: None,
            error: None,
            params: None,
            id: None,
        }
    }
}

struct PackManager
{
    buffer: Vec<u8>,
    cache: [u8; 1024],
    packet: Vec<JsonRpcPacket>,
}

impl PackManager
{
    pub fn new() -> PackManager
    {
        PackManager { buffer: vec![], cache: [0u8;1024], packet: vec![] }
    }

    pub fn append(&mut self, append: &Vec<u8>)
    {
        self.buffer.extend_from_slice(append);
    }

    pub fn parse(&mut self) -> anyhow::Result<bool>
    {
        let mut packet_num = 0;
        let mut start_index = 0;
        while start_index < self.buffer.len() {
            if let Some(index) = self.buffer[start_index..].iter().position(|&byte| byte == b'\n') {
                let end_index = start_index + index;

                let request_str = String::from_utf8_lossy(&self.buffer[start_index..end_index]);
                let request = serde_json::from_str(&request_str)?;

                self.packet.push(request);
                packet_num += 1;

                // 更新 start_index 为换行符之后的位置
                start_index = end_index + 1;
            } else {
                // 没有找到换行符，退出循环
                break;
            }
        }

        self.buffer.drain(0..start_index);
        Ok(packet_num > 0)
    }
}


async fn read_stream_to_buffer(stream: &mut TcpStream) -> anyhow::Result<Vec<u8>>
{
    let mut ret = vec![];
    let mut cache = [0; 1024];
    let mut read_n = 0;
    loop {
        let n = stream.read(&mut cache).await?;
        if n == 0 {
            return bail!("connect close");
        }
        else if n != 0 {
            ret.extend_from_slice(&cache[0..n]);
            read_n += n;
            if stream.readable().await.is_ok() {
                break;
            }
        }
    }
    Ok(ret)
}
pub struct RpcServerConnect
{
    stream: TcpStream,
    pack: PackManager,
    shutdown: bool,
}

impl RpcServerConnect
{
    pub fn new(stream: TcpStream) -> RpcServerConnect
    {
        RpcServerConnect{
            stream: stream,
            pack: PackManager::new(),
            shutdown: false,
        }
    }
}

#[async_trait::async_trait]
trait RpcServerImpl : Send
{
    async fn on_connect(&mut self, rpc:&mut RpcServer,net_id:i32);
    async fn on_disconnect(&mut self, rpc:&mut RpcServer,net_id:i32);
    async fn on_packet(&mut self,rpc:&mut RpcServer,net_id:i32,req:&JsonRpcPacket);
}

struct RpcServerImplEmpty
{
}

#[async_trait::async_trait]
impl RpcServerImpl for RpcServerImplEmpty
{
    async fn on_connect(&mut self, rpc:&mut RpcServer,net_id:i32){}
    async fn on_disconnect(&mut self, rpc:&mut RpcServer,net_id:i32){}
    async fn on_packet(&mut self, rpc:&mut RpcServer,net_id:i32,req:&JsonRpcPacket){}
}

pub struct RpcServer
{
    next_net_id: i32,
    connections: HashMap<i32,RpcServerConnect>,
    shutdown: bool,
}

impl RpcServer
{
    pub fn new() -> RpcServer {
        RpcServer{
            next_net_id: 0,
            connections: Default::default(),
            shutdown: false,
        }
    }

    pub async fn run(server: Arc<Mutex<RpcServer>>, callback: Arc<Mutex<Box<dyn RpcServerImpl>>>, port:i32) -> anyhow::Result<()>
    {
        let url = format!("0.0.0.0:{}",port);
        let listener = TcpListener::bind(url).await?;
        server.lock().await.shutdown = false;

        let mut loop_tick = tokio::time::interval(Duration::from_millis(50));
        loop {
            tokio::select! {
                res = listener.accept() => {
                    let (stream, addr) = res?;

                    let close_handle = server.clone();

                    let mut net_id = -1;
                    //accept
                    {
                        let mut ptr = server.lock().await;
                        net_id = ptr.next_net_id;
                        ptr.next_net_id += 1;

                        let data = RpcServerConnect::new(stream);
                        ptr.connections.insert(net_id, data);
                        debug!("new connect net_id = {}",net_id);

                        let callback_connect = callback.clone();
                        {
                            let mut this = callback_connect.lock().await;
                            this.on_connect(&mut ptr,net_id).await;
                        }
                    }

                    //loop
                    let callback_handle = callback.clone();
                    let bind_handle = server.clone();
                    tokio::spawn(async move {
                        loop {
                            let mut loop_tick = tokio::time::interval(Duration::from_millis(50));
                            {
                                let mut server = bind_handle.lock().await;
                                if server.shutdown {
                                    break;
                                }

                                let ret = server.process_connection(callback_handle.clone(), net_id).await;
                                if ret.is_err() {
                                    debug!("closing connection {}, reason {}",net_id, ret.unwrap_err().to_string());
                                    break;
                                }
                            }
                            loop_tick.tick().await;
                        }

                        //close
                        {
                            let mut ptr = close_handle.lock().await;
                            callback_handle.lock().await.on_disconnect(&mut ptr,net_id).await;
                            ptr.connections.remove(&net_id);
                        }
                    });
                }
                _ = loop_tick.tick() => {
                    let s = server.lock().await;
                    if s.shutdown {
                        println!("shutdown server");
                        return Ok(());
                    }
                }
            }
        }
    }

    pub async fn close(server: Arc<Mutex<RpcServer>>)
    {
        let mut s = server.lock().await;
        s.shutdown = true;
    }

    pub async fn close_stream(&mut self,net_id:i32) -> anyhow::Result<()>
    {
        let d = self.connections.get_mut(&net_id).ok_or(anyhow!("no connection found"))?;
        d.shutdown = true;
        Ok(())
    }

    pub async fn process_connection(&mut self,callback: Arc<Mutex<Box<dyn RpcServerImpl>>>, net_id:i32) -> anyhow::Result<()>
    {
        let mut packets = vec![];
        {
            let d = self.connections.get_mut(&net_id).ok_or(anyhow!("no connection found"))?;
            if d.shutdown {
                return Err(anyhow!("user close"));
            }

            let buffer = read_stream_to_buffer(&mut d.stream).await?;
            d.pack.append(&buffer);
            let parse_ret = d.pack.parse()?;
            if parse_ret == false {
                return Ok(());
            }
            packets.append(&mut d.pack.packet);
        }

        let mut this = callback.lock().await;
        while packets.len() > 0 {
            let pack = packets.pop().unwrap();
            this.on_packet(self,net_id,&pack).await;
        }
        Ok(())
    }

    // 发送消息
    pub async fn send_to(&mut self, net_id:i32, res: &JsonRpcPacket) -> anyhow::Result<()>
    {
        let mut message = serde_json::to_string(&res).unwrap();
        message += "\n";

        let mut d = self.connections.get_mut(&net_id).ok_or(anyhow!("no connection found"))?;
        let _ = d.stream.write_all(message.as_bytes()).await;
        Ok(())
    }

    // 向所有连接发送消息
    pub async fn send_to_all(&mut self, res: &JsonRpcPacket) -> anyhow::Result<()> {
        let mut message = serde_json::to_string(&res).unwrap();
        message += "\n";

        for d in self.connections.values_mut() {
            let _ = d.stream.write_all(message.as_bytes()).await;
        }
        Ok(())
    }
}

pub struct RpcClient
{
    stream: TcpStream,
    pack_manager: PackManager,
}

impl RpcClient
{
    pub async fn new(url: &str) -> anyhow::Result<Self>
    {
        Ok(RpcClient{
            stream: TcpStream::connect(url).await?,
            pack_manager: PackManager::new(),
        })
    }

    pub async fn poll(&mut self) -> anyhow::Result<Vec<JsonRpcPacket>>
    {
        let mut s = &self.stream;
        let ret = self.pack_manager.parse()?;
        if ret == false {
            let ret = self.pack_manager.packet.clone();
            Ok(ret)
        }
        else {
            Ok(Vec::new())
        }
    }

    pub async fn send_no_response(&mut self, request: &JsonRpcPacket) -> anyhow::Result<()>
    {
        let buffer = read_stream_to_buffer(&mut self.stream).await?;
        self.pack_manager.append(&buffer);

        let mut request_json = serde_json::to_string(&request)?;
        request_json += "\n";
        // 发送请求数据
        let ret = self.stream.write_all(request_json.as_bytes()).await?;
        Ok(())
    }

    pub async fn send(&mut self, request: &JsonRpcPacket) -> anyhow::Result<JsonRpcPacket>
    {
        let mut s = &mut self.stream;
        let mut request_json = serde_json::to_string(&request)?;
        request_json += "\n";
        // 发送请求数据
        s.write_all(request_json.as_bytes()).await?;

        let mut interval = interval(Duration::from_millis(50));
        let mut time = 0;
        // 读取响应数据
        loop {
            let buffer = read_stream_to_buffer(&mut s).await?;
            self.pack_manager.append(&buffer);

            let ret = self.pack_manager.parse()?;
            if ret {
                return Ok(self.pack_manager.packet.pop().unwrap());
            }
            else {
                interval.tick().await;
                time += 1;
                if time > 100 {
                    return Err(anyhow::anyhow!("time limit exceeded"));
                }
            }
        }
    }
}

#[cfg(test)]
mod test_server {
    use std::cell::RefCell;
    use rand::random;
    use super::*;

    struct TestServer{
    }

    impl TestServer{
        fn new() -> TestServer {
            TestServer{
            }
        }
    }

    #[async_trait::async_trait]
    impl RpcServerImpl for TestServer
    {
        async fn on_connect(&mut self,rpc:&mut RpcServer,net_id:i32)
        {
            println!("client connected {}",net_id);
        }

        async fn on_disconnect(&mut self,rpc:&mut RpcServer,net_id:i32)
        {
            println!("client disconnected {}",net_id);
        }

        async fn on_packet(&mut self,rpc:&mut RpcServer,net_id:i32,req:&JsonRpcPacket)
        {
            println!("RPC server received packet from {}: {:?}", net_id, req);
            let _ = rpc.send_to(net_id,req).await;
        }
    }

    async fn server_task()
    {
        println!("server_task");

        let mut server = RpcServer::new();
        let test_server = TestServer::new();
        let box_server = Box::new(test_server);
        let box_dyn: Box<dyn RpcServerImpl> = box_server;

        let test_server = Arc::new(Mutex::new(box_dyn));
        let server = Arc::new(Mutex::new(server));
        let server_shutdown = server.clone();
        tokio::spawn(async move {

            let mut inter = interval(Duration::from_secs(10));
            inter.reset();
            inter.tick().await;

            RpcServer::close(server_shutdown).await;
        });

        RpcServer::run(server,test_server,9959).await.unwrap();
    }

    async fn client_task()
    {
        println!("client_task");

        // 异步任务，模拟定时回调
        for i in 0..10 {
            tokio::spawn(async move {

                let mut tk1 = interval(Duration::from_secs(random::<u64>()%10 + 1));
                tk1.reset();
                tk1.tick().await;

                println!("starting client {}", i);
                let mut c = RpcClient::new("127.0.0.1:9959").await.unwrap();

                let mut req = JsonRpcPacket::new();
                req.method = Some("mining.subscribe".to_string());
                let res = c.send(&req).await.unwrap();
                println!("{} -> {:?}", req.method.unwrap(), res);
            });
        }
    }

    #[test]
    fn it_works()
    {
        println!("start rpc test");
        // Initialize the runtime configuration.
        let runtime = runtime::Builder::new_multi_thread()
            .enable_all()
            .thread_stack_size(8 * 1024 * 1024)
            .worker_threads(8)
            .max_blocking_threads(4)
            .build()
            .expect("Failed to initialize a runtime for the router");

        // Initialize the runtime.
        runtime.block_on(async move {

            //client
            client_task().await;

            //server
            server_task().await;

            // Note: Do not move this. The pending await must be here otherwise
            // other snarkOS commands will not exit.
        });
    }
}